package com.chis.jmdatatimer.yfjz.threads;

import com.chis.jm.bean.TriggerBean;
import com.chis.jmdatadb.service.IDataSynRemote;
import com.chis.jmdatatimer.comm.JedisConnect;
import com.chis.jmdatatimer.comm.ServiceConsumer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import redis.clients.jedis.ShardedJedis;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * 
 * @Description 
 * @Company zwx
 * @author wlj
 * @version 2019年01月10日
 */
public class KafkaSrvThread implements Runnable, Serializable {
    private Properties properties;
    private TopicPartition topicPartition;
    private Long offset;
    private String groupid;


    public KafkaSrvThread(Properties properties, TopicPartition topicPartition, Long offset, String groupid) {
        this.properties = properties;
        this.topicPartition = topicPartition;
        this.offset = offset;
        this.groupid = groupid;
    }

    @Override
    public void run() {
        Consumer consumer = new KafkaConsumer(properties);
        consumer.assign(Arrays.asList(topicPartition));
        consumer.seek(topicPartition, offset);
        while (true) {
            ConsumerRecords<String, Object> records = consumer.poll(1000);
            ConsumerRecord<String, Object> lastrecode = null;
            List<TriggerBean> triggerBeanList = null;
            for (ConsumerRecord<String, Object> record : records){
                System.out.printf("partition = %d, offset = %d", record.partition(), record.offset());
                TriggerBean bean = (TriggerBean) record.value();
                triggerBeanList = new ArrayList<>();
                triggerBeanList.add(bean);
                IDataSynRemote remote = (IDataSynRemote) ServiceConsumer.getInstance().getPlugin("dataSynService");
                remote.srvSynService(triggerBeanList);
                lastrecode = record;
            }
            if(lastrecode != null){
                //保存偏移量
                ShardedJedis jedis = null;
                try {
                    jedis = JedisConnect.getInstance().jedisPool.getResource();
                    jedis.select(0);
                    jedis.set(groupid+"-"+lastrecode.topic()+"-"+lastrecode.partition(),String.valueOf(lastrecode.offset()+1));
                } catch (Exception e){
                    e.printStackTrace();
                } finally {
                    if(jedis != null){
                        jedis.close();
                    }
                }
            }

        }
    }
}
